package com.flink.hbase.windowingoptimization.window;

import com.flink.hbase.windowingoptimization.config.WatermarkConfig;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.evictors.TimeEvictor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.ContinuousEventTimeTrigger;
import org.apache.flink.streaming.api.windowing.triggers.CountTrigger;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.time.format.DateTimeFormatter;
import java.util.*;

/**
 * 高阶窗口函数统计技巧
 * 展示生产环境中窗口函数的高级使用方法
 * 
 * 包含功能：
 * 1. 多层级窗口统计
 * 2. 增量聚合与全量聚合结合
 * 3. 自定义触发器和驱逐器
 * 4. 窗口状态管理
 * 5. 复杂业务指标计算
 * 6. 窗口性能优化
 * 
 * @author Flink Developer
 */
public class AdvancedWindowFunctions {
    
    private static final Logger logger = LoggerFactory.getLogger(AdvancedWindowFunctions.class);
    
    /**
     * 交易事件数据模型
     */
    public static class TransactionEvent {
        public String transactionId;
        public String userId;
        public String merchantId;
        public String category;
        public double amount;
        public String currency;
        public long timestamp;
        public String riskLevel;
        public String paymentMethod;
        public String location;
        
        public TransactionEvent() {}
        
        public TransactionEvent(String transactionId, String userId, String merchantId, 
                              String category, double amount, String currency, 
                              long timestamp, String riskLevel, String paymentMethod, 
                              String location) {
            this.transactionId = transactionId;
            this.userId = userId;
            this.merchantId = merchantId;
            this.category = category;
            this.amount = amount;
            this.currency = currency;
            this.timestamp = timestamp;
            this.riskLevel = riskLevel;
            this.paymentMethod = paymentMethod;
            this.location = location;
        }
        
        @Override
        public String toString() {
            return String.format("TransactionEvent{id='%s', userId='%s', merchantId='%s', " +
                    "category='%s', amount=%.2f, currency='%s', timestamp=%d, " +
                    "riskLevel='%s', paymentMethod='%s', location='%s'}",
                    transactionId, userId, merchantId, category, amount, currency, 
                    timestamp, riskLevel, paymentMethod, location);
        }
    }
    
    /**
     * 高级交易统计结果
     */
    public static class AdvancedTransactionStats {
        public String key;
        public long windowStart;
        public long windowEnd;
        public long transactionCount;
        public double totalAmount;
        public double avgAmount;
        public double minAmount;
        public double maxAmount;
        public double stdDev;
        public Set<String> uniqueUsers;
        public Set<String> uniqueMerchants;
        public Map<String, Long> categoryDistribution;
        public Map<String, Long> riskLevelDistribution;
        public Map<String, Double> currencyAmounts;
        public double concentrationIndex;  // 交易集中度指数
        public List<String> topTransactions;
        
        public AdvancedTransactionStats() {
            this.uniqueUsers = new HashSet<>();
            this.uniqueMerchants = new HashSet<>();
            this.categoryDistribution = new HashMap<>();
            this.riskLevelDistribution = new HashMap<>();
            this.currencyAmounts = new HashMap<>();
            this.topTransactions = new ArrayList<>();
        }
        
        @Override
        public String toString() {
            return String.format("AdvancedTransactionStats{key='%s', window=[%s, %s], " +
                    "count=%d, totalAmount=%.2f, avgAmount=%.2f, minAmount=%.2f, " +
                    "maxAmount=%.2f, stdDev=%.2f, uniqueUsers=%d, uniqueMerchants=%d, " +
                    "concentrationIndex=%.4f}",
                    key, formatTimestamp(windowStart), formatTimestamp(windowEnd),
                    transactionCount, totalAmount, avgAmount, minAmount, maxAmount,
                    stdDev, uniqueUsers.size(), uniqueMerchants.size(), concentrationIndex);
        }
        
        private String formatTimestamp(long timestamp) {
            return LocalDateTime.ofInstant(Instant.ofEpochMilli(timestamp), ZoneOffset.UTC)
                    .format(DateTimeFormatter.ISO_LOCAL_DATE_TIME);
        }
    }
    
    /**
     * 高级聚合函数 - 计算复杂统计指标
     */
    public static class AdvancedTransactionAggregateFunction 
            implements AggregateFunction<TransactionEvent, TransactionAccumulator, AdvancedTransactionStats> {
        
        @Override
        public TransactionAccumulator createAccumulator() {
            return new TransactionAccumulator();
        }
        
        @Override
        public TransactionAccumulator add(TransactionEvent transaction, TransactionAccumulator acc) {
            acc.transactionCount++;
            acc.totalAmount += transaction.amount;
            acc.amounts.add(transaction.amount);
            
            // 更新最小值和最大值
            acc.minAmount = Math.min(acc.minAmount, transaction.amount);
            acc.maxAmount = Math.max(acc.maxAmount, transaction.amount);
            
            // 收集唯一用户和商户
            acc.uniqueUsers.add(transaction.userId);
            acc.uniqueMerchants.add(transaction.merchantId);
            
            // 分类分布统计
            acc.categoryDistribution.merge(transaction.category, 1L, Long::sum);
            acc.riskLevelDistribution.merge(transaction.riskLevel, 1L, Long::sum);
            acc.currencyAmounts.merge(transaction.currency, transaction.amount, Double::sum);
            
            // 保存大额交易
            if (transaction.amount > 1000) {
                acc.topTransactions.add(transaction.transactionId + ":" + transaction.amount);
                if (acc.topTransactions.size() > 10) {
                    acc.topTransactions.sort((a, b) -> {
                        double amountA = Double.parseDouble(a.split(":")[1]);
                        double amountB = Double.parseDouble(b.split(":")[1]);
                        return Double.compare(amountB, amountA);
                    });
                    acc.topTransactions = acc.topTransactions.subList(0, 10);
                }
            }
            
            return acc;
        }
        
        @Override
        public AdvancedTransactionStats getResult(TransactionAccumulator acc) {
            AdvancedTransactionStats stats = new AdvancedTransactionStats();
            
            stats.transactionCount = acc.transactionCount;
            stats.totalAmount = acc.totalAmount;
            stats.avgAmount = acc.transactionCount > 0 ? acc.totalAmount / acc.transactionCount : 0;
            stats.minAmount = acc.minAmount;
            stats.maxAmount = acc.maxAmount;
            
            // 计算标准差
            if (acc.transactionCount > 1) {
                double variance = 0;
                for (double amount : acc.amounts) {
                    variance += Math.pow(amount - stats.avgAmount, 2);
                }
                stats.stdDev = Math.sqrt(variance / (acc.transactionCount - 1));
            }
            
            stats.uniqueUsers = acc.uniqueUsers;
            stats.uniqueMerchants = acc.uniqueMerchants;
            stats.categoryDistribution = acc.categoryDistribution;
            stats.riskLevelDistribution = acc.riskLevelDistribution;
            stats.currencyAmounts = acc.currencyAmounts;
            stats.topTransactions = acc.topTransactions;
            
            // 计算集中度指数 (Herfindahl-Hirschman Index)
            stats.concentrationIndex = calculateConcentrationIndex(acc.categoryDistribution, acc.transactionCount);
            
            return stats;
        }
        
        @Override
        public TransactionAccumulator merge(TransactionAccumulator acc1, TransactionAccumulator acc2) {
            acc1.transactionCount += acc2.transactionCount;
            acc1.totalAmount += acc2.totalAmount;
            acc1.amounts.addAll(acc2.amounts);
            
            acc1.minAmount = Math.min(acc1.minAmount, acc2.minAmount);
            acc1.maxAmount = Math.max(acc1.maxAmount, acc2.maxAmount);
            
            acc1.uniqueUsers.addAll(acc2.uniqueUsers);
            acc1.uniqueMerchants.addAll(acc2.uniqueMerchants);
            
            // 合并分布统计
            acc2.categoryDistribution.forEach((k, v) -> acc1.categoryDistribution.merge(k, v, Long::sum));
            acc2.riskLevelDistribution.forEach((k, v) -> acc1.riskLevelDistribution.merge(k, v, Long::sum));
            acc2.currencyAmounts.forEach((k, v) -> acc1.currencyAmounts.merge(k, v, Double::sum));
            
            // 合并大额交易
            acc1.topTransactions.addAll(acc2.topTransactions);
            if (acc1.topTransactions.size() > 10) {
                acc1.topTransactions.sort((a, b) -> {
                    double amountA = Double.parseDouble(a.split(":")[1]);
                    double amountB = Double.parseDouble(b.split(":")[1]);
                    return Double.compare(amountB, amountA);
                });
                acc1.topTransactions = acc1.topTransactions.subList(0, 10);
            }
            
            return acc1;
        }
        
        private double calculateConcentrationIndex(Map<String, Long> distribution, long total) {
            if (total == 0) return 0;
            
            double hhi = 0;
            for (long count : distribution.values()) {
                double share = (double) count / total;
                hhi += share * share;
            }
            return hhi;
        }
    }
    
    /**
     * 交易聚合器
     */
    public static class TransactionAccumulator {
        public long transactionCount = 0;
        public double totalAmount = 0;
        public double minAmount = Double.MAX_VALUE;
        public double maxAmount = Double.MIN_VALUE;
        public List<Double> amounts = new ArrayList<>();
        public Set<String> uniqueUsers = new HashSet<>();
        public Set<String> uniqueMerchants = new HashSet<>();
        public Map<String, Long> categoryDistribution = new HashMap<>();
        public Map<String, Long> riskLevelDistribution = new HashMap<>();
        public Map<String, Double> currencyAmounts = new HashMap<>();
        public List<String> topTransactions = new ArrayList<>();
    }
    
    /**
     * 增量聚合 + 全量处理的窗口函数
     */
    public static class IncrementalWithFullWindowFunction 
            extends ProcessWindowFunction<AdvancedTransactionStats, AdvancedTransactionStats, String, TimeWindow> {
        
        private transient ValueState<AdvancedTransactionStats> previousWindowState;
        
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            
            ValueStateDescriptor<AdvancedTransactionStats> descriptor = 
                new ValueStateDescriptor<>("previous-window-stats", AdvancedTransactionStats.class);
            previousWindowState = getRuntimeContext().getState(descriptor);
        }
        
        @Override
        public void process(String key, Context context, 
                          Iterable<AdvancedTransactionStats> elements, 
                          Collector<AdvancedTransactionStats> out) throws Exception {
            
            AdvancedTransactionStats current = elements.iterator().next();
            current.key = key;
            current.windowStart = context.window().getStart();
            current.windowEnd = context.window().getEnd();
            
            // 获取上一个窗口的统计数据
            AdvancedTransactionStats previous = previousWindowState.value();
            
            // 计算趋势信息
            if (previous != null) {
                calculateTrends(current, previous);
            }
            
            // 异常检测
            if (detectAnomalies(current, previous)) {
                logger.warn("检测到异常交易模式: {}", current);
            }
            
            // 更新状态
            previousWindowState.update(current);
            
            out.collect(current);
        }
        
        private void calculateTrends(AdvancedTransactionStats current, AdvancedTransactionStats previous) {
            // 计算增长率
            double countGrowth = calculateGrowthRate(current.transactionCount, previous.transactionCount);
            double amountGrowth = calculateGrowthRate(current.totalAmount, previous.totalAmount);
            
            logger.info("趋势分析 [{}]: 交易量增长率 {:.2f}%, 交易金额增长率 {:.2f}%",
                current.key, countGrowth * 100, amountGrowth * 100);
        }
        
        private double calculateGrowthRate(double current, double previous) {
            if (previous == 0) return current > 0 ? 1.0 : 0.0;
            return (current - previous) / previous;
        }
        
        private boolean detectAnomalies(AdvancedTransactionStats current, AdvancedTransactionStats previous) {
            if (previous == null) return false;
            
            // 异常检测规则
            
            // 1. 交易量突然增加超过200%
            if (current.transactionCount > previous.transactionCount * 3) {
                return true;
            }
            
            // 2. 交易金额突然增加超过500%
            if (current.totalAmount > previous.totalAmount * 5) {
                return true;
            }
            
            // 3. 高风险交易比例超过50%
            Long highRiskCount = current.riskLevelDistribution.getOrDefault("HIGH", 0L);
            if (current.transactionCount > 0 && 
                (double) highRiskCount / current.transactionCount > 0.5) {
                return true;
            }
            
            return false;
        }
    }
    
    /**
     * 多层级窗口统计示例
     */
    public void runMultiLayerWindowAnalysis() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        
        // 模拟数据源
        DataStream<TransactionEvent> transactionStream = env.addSource(new TransactionEventSource());
        
        // 设置Watermark
        SingleOutputStreamOperator<TransactionEvent> watermarkedStream = transactionStream
            .assignTimestampsAndWatermarks(
                WatermarkConfig.mixedScenarioStrategy(
                    (event, recordTimestamp) -> event.timestamp
                )
            );
        
        // 第一层：1分钟窗口实时统计
        SingleOutputStreamOperator<AdvancedTransactionStats> minuteStats = watermarkedStream
            .keyBy(event -> event.category)
            .window(TumblingEventTimeWindows.of(Time.minutes(1)))
            .aggregate(
                new AdvancedTransactionAggregateFunction(),
                new IncrementalWithFullWindowFunction()
            )
            .name("minute-level-stats");
        
        // 第二层：5分钟滑动窗口趋势分析
        SingleOutputStreamOperator<AdvancedTransactionStats> trendStats = watermarkedStream
            .keyBy(event -> event.category)
            .window(SlidingEventTimeWindows.of(Time.minutes(5), Time.minutes(1)))
            .aggregate(
                new AdvancedTransactionAggregateFunction(),
                new TrendAnalysisWindowFunction()
            )
            .name("trend-analysis");
        
        // 第三层：15分钟全局统计
        SingleOutputStreamOperator<AdvancedTransactionStats> globalStats = watermarkedStream
            .windowAll(TumblingEventTimeWindows.of(Time.minutes(15)))
            .aggregate(
                new GlobalTransactionAggregateFunction(),
                new GlobalAnalysisWindowFunction()
            )
            .name("global-analysis");
        
        // 输出结果
        minuteStats.print("分钟级统计");
        trendStats.print("趋势分析");
        globalStats.print("全局统计");
        
        env.execute("Multi-Layer Window Analysis");
    }
    
    /**
     * 趋势分析窗口函数
     */
    public static class TrendAnalysisWindowFunction 
            extends ProcessWindowFunction<AdvancedTransactionStats, AdvancedTransactionStats, String, TimeWindow> {
        
        @Override
        public void process(String key, Context context, 
                          Iterable<AdvancedTransactionStats> elements, 
                          Collector<AdvancedTransactionStats> out) throws Exception {
            
            AdvancedTransactionStats stats = elements.iterator().next();
            stats.key = key + "_trend";
            stats.windowStart = context.window().getStart();
            stats.windowEnd = context.window().getEnd();
            
            // 趋势分析逻辑
            analyzeTrends(stats);
            
            out.collect(stats);
        }
        
        private void analyzeTrends(AdvancedTransactionStats stats) {
            // 计算集中度变化
            if (stats.concentrationIndex > 0.5) {
                logger.info("检测到高集中度交易模式: {} (集中度指数: {:.4f})", 
                    stats.key, stats.concentrationIndex);
            }
            
            // 分析风险分布
            analyzeRiskDistribution(stats);
        }
        
        private void analyzeRiskDistribution(AdvancedTransactionStats stats) {
            long totalTransactions = stats.transactionCount;
            if (totalTransactions > 0) {
                for (Map.Entry<String, Long> entry : stats.riskLevelDistribution.entrySet()) {
                    double percentage = (double) entry.getValue() / totalTransactions * 100;
                    logger.info("风险级别 {} 占比: {:.2f}%", entry.getKey(), percentage);
                }
            }
        }
    }
    
    /**
     * 全局聚合函数
     */
    public static class GlobalTransactionAggregateFunction 
            implements AggregateFunction<TransactionEvent, GlobalTransactionAccumulator, AdvancedTransactionStats> {
        
        @Override
        public GlobalTransactionAccumulator createAccumulator() {
            return new GlobalTransactionAccumulator();
        }
        
        @Override
        public GlobalTransactionAccumulator add(TransactionEvent transaction, GlobalTransactionAccumulator acc) {
            acc.totalTransactions++;
            acc.totalAmount += transaction.amount;
            
            // 按类别统计
            acc.categoryStats.computeIfAbsent(transaction.category, k -> new CategoryStats())
                .addTransaction(transaction);
            
            // 按风险级别统计
            acc.riskStats.merge(transaction.riskLevel, 1L, Long::sum);
            
            // 按地区统计
            acc.locationStats.merge(transaction.location, transaction.amount, Double::sum);
            
            return acc;
        }
        
        @Override
        public AdvancedTransactionStats getResult(GlobalTransactionAccumulator acc) {
            AdvancedTransactionStats stats = new AdvancedTransactionStats();
            
            stats.transactionCount = acc.totalTransactions;
            stats.totalAmount = acc.totalAmount;
            stats.avgAmount = acc.totalTransactions > 0 ? acc.totalAmount / acc.totalTransactions : 0;
            
            // 计算全局指标
            stats.categoryDistribution = acc.categoryStats.entrySet().stream()
                .collect(HashMap::new, 
                    (map, entry) -> map.put(entry.getKey(), entry.getValue().transactionCount),
                    (map1, map2) -> map1.putAll(map2));
            
            stats.riskLevelDistribution = acc.riskStats;
            
            return stats;
        }
        
        @Override
        public GlobalTransactionAccumulator merge(GlobalTransactionAccumulator acc1, 
                                                 GlobalTransactionAccumulator acc2) {
            acc1.totalTransactions += acc2.totalTransactions;
            acc1.totalAmount += acc2.totalAmount;
            
            // 合并类别统计
            acc2.categoryStats.forEach((category, stats) -> 
                acc1.categoryStats.merge(category, stats, CategoryStats::merge));
            
            // 合并风险统计
            acc2.riskStats.forEach((risk, count) -> 
                acc1.riskStats.merge(risk, count, Long::sum));
            
            // 合并地区统计
            acc2.locationStats.forEach((location, amount) -> 
                acc1.locationStats.merge(location, amount, Double::sum));
            
            return acc1;
        }
    }
    
    /**
     * 全局交易累加器
     */
    public static class GlobalTransactionAccumulator {
        public long totalTransactions = 0;
        public double totalAmount = 0;
        public Map<String, CategoryStats> categoryStats = new HashMap<>();
        public Map<String, Long> riskStats = new HashMap<>();
        public Map<String, Double> locationStats = new HashMap<>();
    }
    
    /**
     * 类别统计
     */
    public static class CategoryStats {
        public long transactionCount = 0;
        public double totalAmount = 0;
        public double avgAmount = 0;
        
        public void addTransaction(TransactionEvent transaction) {
            transactionCount++;
            totalAmount += transaction.amount;
            avgAmount = totalAmount / transactionCount;
        }
        
        public static CategoryStats merge(CategoryStats stats1, CategoryStats stats2) {
            CategoryStats merged = new CategoryStats();
            merged.transactionCount = stats1.transactionCount + stats2.transactionCount;
            merged.totalAmount = stats1.totalAmount + stats2.totalAmount;
            merged.avgAmount = merged.transactionCount > 0 ? 
                merged.totalAmount / merged.transactionCount : 0;
            return merged;
        }
    }
    
    /**
     * 全局分析窗口函数
     */
    public static class GlobalAnalysisWindowFunction 
            extends ProcessAllWindowFunction<AdvancedTransactionStats, AdvancedTransactionStats, TimeWindow> {
        
        @Override
        public void process(Context context, 
                          Iterable<AdvancedTransactionStats> elements, 
                          Collector<AdvancedTransactionStats> out) throws Exception {
            
            AdvancedTransactionStats stats = elements.iterator().next();
            stats.key = "global";
            stats.windowStart = context.window().getStart();
            stats.windowEnd = context.window().getEnd();
            
            // 全局分析
            performGlobalAnalysis(stats);
            
            out.collect(stats);
        }
        
        private void performGlobalAnalysis(AdvancedTransactionStats stats) {
            logger.info("全局分析结果: 总交易数={}, 总金额={:.2f}, 平均金额={:.2f}",
                stats.transactionCount, stats.totalAmount, stats.avgAmount);
            
            // 输出top类别
            stats.categoryDistribution.entrySet().stream()
                .sorted(Map.Entry.<String, Long>comparingByValue().reversed())
                .limit(5)
                .forEach(entry -> 
                    logger.info("热门类别: {} ({}笔交易)", entry.getKey(), entry.getValue()));
        }
    }
    
    /**
     * 自定义触发器示例：基于事件数量和时间的混合触发
     */
    public void runCustomTriggerExample() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        
        DataStream<TransactionEvent> transactionStream = env.addSource(new TransactionEventSource());
        
        SingleOutputStreamOperator<TransactionEvent> watermarkedStream = transactionStream
            .assignTimestampsAndWatermarks(
                WatermarkConfig.lowLatencyStrategy(
                    (event, recordTimestamp) -> event.timestamp
                )
            );
        
        // 使用自定义触发器：每100笔交易或每30秒触发一次
        SingleOutputStreamOperator<AdvancedTransactionStats> customTriggerStats = watermarkedStream
            .keyBy(event -> event.merchantId)
            .window(TumblingEventTimeWindows.of(Time.minutes(5)))
            .trigger(CountTrigger.of(100))  // 每100笔交易触发
            .evictor(TimeEvictor.of(Time.seconds(30)))  // 保留最近30秒的数据
            .aggregate(
                new AdvancedTransactionAggregateFunction(),
                new CustomTriggerWindowFunction()
            )
            .name("custom-trigger-stats");
        
        customTriggerStats.print("自定义触发器统计");
        
        env.execute("Custom Trigger Example");
    }
    
    /**
     * 自定义触发器窗口函数
     */
    public static class CustomTriggerWindowFunction 
            extends ProcessWindowFunction<AdvancedTransactionStats, AdvancedTransactionStats, String, TimeWindow> {
        
        @Override
        public void process(String key, Context context, 
                          Iterable<AdvancedTransactionStats> elements, 
                          Collector<AdvancedTransactionStats> out) throws Exception {
            
            AdvancedTransactionStats stats = elements.iterator().next();
            stats.key = key + "_custom";
            stats.windowStart = context.window().getStart();
            stats.windowEnd = context.window().getEnd();
            
            // 记录触发信息
            logger.info("自定义触发器触发: 商户={}, 窗口=[{}, {}], 交易数={}",
                key, stats.windowStart, stats.windowEnd, stats.transactionCount);
            
            out.collect(stats);
        }
    }
    
    /**
     * 模拟交易事件源
     */
    public static class TransactionEventSource 
            extends org.apache.flink.streaming.api.functions.source.RichSourceFunction<TransactionEvent> {
        
        private volatile boolean running = true;
        private final Random random = new Random();
        
        @Override
        public void run(SourceContext<TransactionEvent> ctx) throws Exception {
            while (running) {
                TransactionEvent event = generateRandomTransaction();
                ctx.collect(event);
                Thread.sleep(random.nextInt(100) + 50);  // 50-150ms间隔
            }
        }
        
        @Override
        public void cancel() {
            running = false;
        }
        
        private TransactionEvent generateRandomTransaction() {
            String[] categories = {"食品", "服装", "电子", "家居", "汽车", "娱乐", "医疗", "教育"};
            String[] riskLevels = {"LOW", "MEDIUM", "HIGH"};
            String[] paymentMethods = {"信用卡", "借记卡", "支付宝", "微信", "现金"};
            String[] locations = {"北京", "上海", "广州", "深圳", "杭州", "成都", "武汉", "西安"};
            String[] currencies = {"CNY", "USD", "EUR", "JPY"};
            
            return new TransactionEvent(
                "TXN" + random.nextInt(1000000),
                "USER" + random.nextInt(10000),
                "MERCHANT" + random.nextInt(1000),
                categories[random.nextInt(categories.length)],
                random.nextDouble() * 5000 + 10,
                currencies[random.nextInt(currencies.length)],
                System.currentTimeMillis(),
                riskLevels[random.nextInt(riskLevels.length)],
                paymentMethods[random.nextInt(paymentMethods.length)],
                locations[random.nextInt(locations.length)]
            );
        }
    }
    
    public static void main(String[] args) throws Exception {
        AdvancedWindowFunctions example = new AdvancedWindowFunctions();
        
        // 运行多层级窗口分析
        example.runMultiLayerWindowAnalysis();
        
        // 或者运行自定义触发器示例
        // example.runCustomTriggerExample();
    }
} 